Exercise: Cloud Service

Suppose you have the following class that is a helper class for managing access to a Google Cloud environment:

@dataclass
class CloudService:
  auth_provider: GoogleCredentials
  service: GoogleServiceProvider
  storage_manager: GoogleStorage

  def connect(self) -> None:
    print("Connecting to the cloud service.")
    credentials = self.auth_provider.retrieve_credentials()
    self.service.connect(credentials)
    context = self.service.get_context()
    self.storage_manager.initialize(context)
    print("Cloud service connected.")

As you can see, the class directly depends on Google-specific classes: GoogleCredentials, GoogleService, and GoogleStorage. You want to remove this direct dependency. However, you can't change the original classes provided by Google. In fact, you don't even have access to the source code of those classes. How do you solve this? Refactor your code to remove the direct dependency.

Compatible Python Versions: 3.8+


Hugi Asgeirsson

I have a hard time understanding the usefulness of this abstraction.
Sure, you can define a protocol, like:

class ServiceProvider(Protocol):
def connect(self, credentials: str):
...

def get_context(self) -> str:
...

What you gain from this, theoretically, is that if you ever want to use another service provider you could pass another service provider _if_ it happened to have functions with exactly the same names that gave the exact same return types. But how likely or common is that really? Not at all likely, I'd say.

Am I missing something? This seems a little contrived.

REPLY
Andreas [ArjanCodes Team]

I cannot say how often or common it is to have multiple cloud providers. But the idea of having this abstraction with the protocol is not to constrain the ServiceProvider but rather to say, "Do I have a certain structure to be used in this context?". This puts some constraints if we define another class that should adhere to the ServiceProvider, but it also creates a contract for other parts of our codebase.

For example:

from typing import Protocol

class ServiceProvider(Protocol):
def connect(self, credentials: str): ...

def get_context(self) -> str: ...

class MyProvider:
def connect(self, credentials: str):
print(f"Connecting to {credentials}")

def get_context(self) -> str:
return "Context"

class MyOtherProvider:
def connect(self, credentials: str):
print(f"Connecting to {credentials}")

def get_context(self) -> str:
return "Context"

def is_available(self) -> bool:
return True

def connect_to_service(provider: ServiceProvider, credentials: str):
provider.connect(credentials)
print(provider.get_context())

connect_to_service(MyProvider(), "my credentials")
connect_to_service(MyOtherProvider(), "my credentials")

connect_to_service expects to have a argument that has the methods connect, get_context.

REPLY
Hugi Asgeirsson

Yes, I get that. But in this case the instruction was:
"However, you can't change the original classes provided by Google. In fact, you don't even have access to the source code of those classes."

This means that what we are dealing with is a third-party library provided by Google. If we had another cloud service provider, say AWS, they would probably have their own SDK with other classes.

What I am saying is that is extremely unlikely that AWSServiceProvider also has a methods named connect and get_context with the same return types. AWSServiceProvider probably has methods that perform the same job, but unless they have the same name and return type, the Protocol is not helpful. Am I missing something?

I'm trying to understand if a protocol is actually useful in a situation like this when dealing with a number of interchangeable third party SDKs, or if it's just an example to demonstrate protocols, in which case there may be better examples.

REPLY
Andreas [ArjanCodes Team]

I get the point that it is not likely that the different SDKs or libraries would match. In this case, exercise, we wanted to show that protocols allow you to establish a consistent contract in your code.

But it should also include for example adapter patterns that adhere to the protocol in which we can pass through dependency injection.

We will look into this exercise and update it accordingly!

Do you have any suggestions/clarifications/improvements that you think would be nice?

REPLY
Hugi Asgeirsson

I would probably just change the background information that assumes the class is something imported from a Google third party library. For example, you could say instead that the cloud provider classes are an interface written by your colleague where you work, which makes it at least plausible that interfaces for other cloud providers in the future would implement the same methods.

However, most of all I would like Arjan to discuss when, if and how you should bother to implement protocols when interacting with third party services. I actually implemented a protocol like this a few months ago in a project, and later removed the protocol when I realized that it was really just an unnecessary abstraction layer because there was no plausible situation real-world situation where I would actually be able to re-use it.

REPLY
ABDALLAH EL HIDALI

Exercice1:
from typing import Protocol
from dataclasses import dataclass

class Credentials(Protocol):
def retrieve_credentials(self) -> str:
...

class ServiceProvider(Protocol):
def connect(self, credentials: str) -> None:
...

def get_context(self) -> str:
...

class StorageManager(Protocol):
def initialize(self, context: str) -> None:
...

@dataclass
class CloudService:
auth_provider: Credentials
service: ServiceProvider
storage_manager: StorageManager

def connect(self) -> None:
print("Connecting to the cloud service.")
credentials = self.auth_provider.retrieve_credentials()
self.service.connect(credentials)
context = self.service.get_context()
self.storage_manager.initialize(context)
print("Cloud service connected.")

def main() -> None:
from google_service import GoogleCredentials, GoogleServiceProvider, GoogleStorage

credentials = GoogleCredentials()
service = GoogleServiceProvider()
storage = GoogleStorage()
cloud_service = CloudService(credentials, service, storage)
cloud_service.connect()

if __name__ == "__main__":
main()

Eercice2:
from typing import Protocol
from dataclasses import dataclass

@dataclass
class SenderSettings:
DEFAULT_EMAIL = "support@arjancodes.com"
LOGIN = "test"
PASSWORD = "my_password"
HOST = "smtp.arjancodes.com"
PORT = 19584

@dataclass
class UserMessage:
email: str
message: str

class EmailServer(Protocol):
def connect(self, host: str, port: int) -> None:
...

def login(self, login: str, password: str) -> None:
...

def sendmail(self, from_email: str, to_email: str, message: str) -> None:
...

def quit(self) -> None:
...

def send_email(
user_message: UserMessage, sender_settings: SenderSettings, server: EmailServer
) -> None:
server.connect(sender_settings.HOST, sender_settings.PORT)
server.login(sender_settings.LOGIN, sender_settings.PASSWORD)
server.sendmail(sender_settings.DEFAULT_EMAIL, user_message.email, user_message.message)
server.quit()

REPLY
Andreas [ArjanCodes Team]

Looks good! Nice solution with removing the dependencies!

REPLY
Alexandre Gourmelon

Exercice 1:

from dataclasses import dataclass
from typing import Protocol

class Credentials(Protocol):

def retrieve_credentials(self) -> str:
...

class Provider(Protocol):

def connect(self, credentials: str) -> None:
...

def get_context(self) -> str:
...

class Storage(Protocol):

def initialize(self, context: str) -> None:
...

@dataclass
class CloudService:
auth_provider: Credentials
service: Provider
storage_manager: Storage

def connect(self) -> None:
print("Connecting to the cloud service.")
credentials = self.auth_provider.retrieve_credentials()
self.service.connect(credentials)
context = self.service.get_context()
self.storage_manager.initialize(context)
print("Cloud service connected.")

Exercice 2:

from typing import Protocol, Any

DEFAULT_EMAIL = "support@arjancodes.com"
LOGIN = "test"
PASSWORD = "my_password"
HOST = "smtp.arjancodes.com"
PORT = 19584

class MailServer(Protocol):

def connect(self, host: str, port: int) -> Any:
...

def login(self, login: str, password: str) -> Any:
...

def sendmail(self, from_address: str, to_address: str, message: str) -> Any:
...

def quit(self) -> Any:
...

def send_email(
server: MailServer, message: str, to_address: str, from_address: str = DEFAULT_EMAIL
) -> None:
server.connect(HOST, PORT)
server.login(LOGIN, PASSWORD)
server.sendmail(from_address, to_address, message)
server.quit()

REPLY
Andreas [ArjanCodes Team]

Nice solution Alexandre, have you used protocols like this before?

REPLY
Agustin Rodriguez

Hey . My solution
eg:1
@dataclass
class CloudService:
auth_provider: GoogleCredentials
service: GoogleServiceProvider
storage_manager: GoogleStorage

def connect(self) -> None:
print("Connecting to the cloud service.")
credentials = self.auth_provider.retrieve_credentials()
self.service.connect(credentials)
context = self.service.get_context()
self.storage_manager.initialize(context)
print("Cloud service connected.")
eg2:
# depens on SMTP

from smtplib import SMTP

DEFAULT_EMAIL = "support@arjancodes.com"
LOGIN = "test"
PASSWORD = "my_password"
HOST = "smtp.arjancodes.com"
PORT = 19584

def SMTProvider(Protocol):
def connect(host:str, port:int)-> None:
...
def login(login:str, password:str)-> None:
...
def sendmail(from_address:str, to_address:str, message:str)-> None:
...
def quit()-> None:
...

def send_email(
mail_server: SMTProvider, message: str, to_address: str, from_address: str = DEFAULT_EMAIL
) -> None:
# server = SMTP()
mail_server.connect(HOST, PORT)
mail_server.login(LOGIN, PASSWORD)
mail_server.sendmail(from_address, to_address, message)
mail_server.quit()

REPLY
Andreas [ArjanCodes Team]

Good start! However, the first exercise does seem incomplete because there are no abstractions introduced. For the second exercise, this looks good! I would however recommend that the protocol gets renamed because it should describe the wanted structure and not the structure needed for only SMTP.

REPLY
Agustin Rodriguez

Sorry, in the first solution I pasted the exercise, not the solution.

from typing import Protocol

class CredentiaslManager(Protocol):
def retrive_credentials(self)-> str:
...

class ServiceProvider(Protocol):
def connect(self, credentials: str)-> None:
...

def get_context(self)-> any:
...

class StorageManager(Protocol):
def initialize(self, context: str)-> None:
...

@dataclass
class CloudService:
auth_provider: CredentiaslManager
service: ServiceProvider
storage_manager: StorageManager
def connect(self) -> None:
print("Connecting to the cloud service.")
credentials = self.auth_provider.retrieve_credentials()

self.service.connect(credentials)
context = self.service.get_context()
self.storage_manager.initialize(context)
print("Cloud service connected.")

REPLY
Agustin Rodriguez

or the second problem, is MailServiceProvider a better name?

REPLY
Andreas [ArjanCodes Team]

No worries! I figured something like that had happend :)

REPLY
Andreas [ArjanCodes Team]

I think something like that is a bit more generic, but also quite specifc!

REPLY
Philipp Walter

Really cool chapter and nice exercise. Especially the last one is an good example for me how this can help to test the code or in my example just try something:

Looking forward to implement Protocol in my daily work :-)!

For those who are interested in the usage of "send_mail" with a dummy "SMTP" (Remark: I created the class "MyEmailTransferProtocol" inside the main to remember myself that this would normaly part of another file, like the test- or main- file -> where send_email is used)):

def send_email(
mtp: EmailTransferProtocol,
message: str,
to_address: str,
from_address: str = DEFAULT_EMAIL,
) -> None:
server = mtp
server.connect(HOST, PORT)
server.login(LOGIN, PASSWORD)
server.sendmail(from_address, to_address, message)
server.quit()

def main() -> None:

class MyEmailTransferProtocol:
def connect(self, host: str, port: int) -> None:
print(f"Connecting to {host} on port {port}.")

def login(self, login: str, password: str) -> None:
print(f"Logging in as {login}.")

def sendmail(self, from_address: str, to_address: str, message: str) -> None:
print(
f"Sending email from {from_address} to {to_address} with message: {message}"
)

def quit(self) -> None:
print("Quitting the email server.")

send_email(
mtp=MyEmailTransferProtocol(),
message="Hello, world!",
to_address="refactored@with.protocol",
)

if __name__ == "__main__":
main()

REPLY
Andreas [ArjanCodes Team]

Nice solution! Protocols are for sure a feature that needs to be used more in Python projects. Makes the code a lot more maintainable and testable.

REPLY
Show More